package com.atguigu.mapreduce.demo.writable;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @version V1.0
 * @author: guangcheng
 * @date: 2021/11/18 23:49
 * @description:
 */
public class PCounterDriver {

    private static String inputPath = "D:\\testarea\\bigdatatest\\hadoop\\mapreduce\\11_input\\inputflow";
    private static String outputPath = "D:\\testarea\\bigdatatest\\hadoop\\mapreduce\\11_input\\inputflow\\outpath";

    public static void main(String[] args) {

        try {

            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);

            job.setJarByClass(PCounterDriver.class);

            job.setMapperClass(PCounterMapper.class);
            job.setReducerClass(PCounterReduce.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(PhoneCounterPojo.class);

            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(PhoneCounterPojo.class);

            //切片设置
            //job.setInputFormatClass(CombineTextInputFormat.class);
            //虚拟存储改大一点, 就会把小文件尽可能逻辑上聚集在一起, 用更少的切片去处理
            //CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);//4M

            //使用自定义的partition
            //job.setPartitionerClass(MyPartitioner.class);
            //设置reduceTask数量
            //job.setNumReduceTasks(5);

            FileInputFormat.setInputPaths(job, new Path(inputPath));
            FileOutputFormat.setOutputPath(job, new Path(outputPath));

            boolean res = job.waitForCompletion(true);
            System.exit(res ? 0 : 1);

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}

